package com.bigdata.warehouse.dws;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: com.bigdata.warehouse.dws
 * @Author: zhangleqing
 * @CreateTime: 2025-08-24  11:34
 * @Description: TODO
 * @Version: 1.0
 * 页面总体统计
 */
public class PageOverallCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 读取Doris中的 community（小区表）
        tEnv.executeSql(
                "CREATE TABLE ods_community (\n" +
                        "    id INT,\n" +
                        "    communityName STRING,\n" +
                        "    district_id INT,\n" +
                        "    city_id INT,\n" +
                        "    communityLocation STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.community',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        // 读取Doris中的 house（楼栋表）
        tEnv.executeSql(
                "CREATE TABLE ods_house (\n" +
                        "    id INT,\n" +
                        "    cid INT,\n" +
                        "    houseCode STRING,\n" +
                        "    houseName STRING,\n" +
                        "    buildStartTime STRING,\n" +
                        "    buildEndTime STRING,\n" +
                        "    houseArea STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.house',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        // 读取Doris中的 ower（业主表）
        tEnv.executeSql(
                "CREATE TABLE ods_ower (\n" +
                        "    id INT,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime STRING,\n" +
                        "    operatorTime STRING\n" +
                        ") WITH (\n" +
                        "'connector' = 'doris',\n" +
                        "'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "'table.identifier' = 'sca_doris.ower',\n" +
                        "'username' = 'root',\n" +
                        "'password' = '123456'\n" +
                        ")"
        );

        // 读取Doris中的 cars（车辆信息表）
        tEnv.executeSql(
                "CREATE TABLE ods_cars (\n" +
                        "    id INT,\n" +
                        "    owerId INT,\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    `type` STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.cars',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );




/*
create table dws_community_wide (
    community_id INT,
    community_name STRING,
    district_id INT,
    city_id INT,
    community_location STRING,
    community_remark STRING,
    house_id INT,
    house_cid INT,
    house_code STRING,
    house_name STRING,
    build_start_time TIMESTAMP(3),
    build_end_time TIMESTAMP(3),
    house_area DECIMAL(10,2),
    house_remark STRING,
    ower_id INT,
    ower_room_id INT,
    ower_code STRING,
    ower_name STRING,
    ower_sex STRING,
    ower_age INT,
    ower_idcard STRING,
    ower_tel STRING,
    ower_homeplace STRING,
    ower_workplace STRING,
    ower_create_time TIMESTAMP(3),
    ower_operator_time TIMESTAMP(3),
    ower_remark STRING,
    car_id INT,
    car_ower_id INT,
    car_code STRING,
    car_color STRING,
    car_type STRING,
    car_remark STRING
) WITH (
    'connector' = 'doris',
    'fenodes' = 'node102:8030',
    'table.identifier' = 'sca_doris_export.dws_community_wide',
    'username' = 'root',
    'password' = '123456',
    'sink.buffer-flush.max-rows' = '1000',  -- 累计1千条数据触发写入
    'sink.buffer-flush.interval' = '5000',   -- 若5秒内未达1千条，也触发写入
    'sink.max-retries' = '3',
    'sink.enable-delete' = 'true',
    'sink.properties.format' = 'json',
    'sink.enable.batch-mode' = 'true'
)
 */     // 创建大宽表
        tEnv.executeSql(
                "create table dws_community_wide (\n" +
                        "    community_id INT,\n" +
                        "    community_name STRING,\n" +
                        "    district_id INT,\n" +
                        "    city_id INT,\n" +
                        "    community_location STRING,\n" +
                        "    community_remark STRING,\n" +
                        "    house_id INT,\n" +
                        "    house_cid INT,\n" +
                        "    house_code STRING,\n" +
                        "    house_name STRING,\n" +
                        "    build_start_time STRING,\n" +
                        "    build_end_time STRING,\n" +
                        "    house_area STRING,\n" +
                        "    house_remark STRING,\n" +
                        "    ower_id INT,\n" +
                        "    ower_room_id INT,\n" +
                        "    ower_code STRING,\n" +
                        "    ower_name STRING,\n" +
                        "    ower_sex STRING,\n" +
                        "    ower_age INT,\n" +
                        "    ower_idcard STRING,\n" +
                        "    ower_tel STRING,\n" +
                        "    ower_homeplace STRING,\n" +
                        "    ower_workplace STRING,\n" +
                        "    ower_create_time STRING,\n" +
                        "    ower_operator_time STRING,\n" +
                        "    ower_remark STRING,\n" +
                        "    car_id INT,\n" +
                        "    car_ower_id INT,\n" +
                        "    car_code STRING,\n" +
                        "    car_color STRING,\n" +
                        "    car_type STRING,\n" +
                        "    car_remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris_export.dws_community_wide',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.buffer-flush.max-rows' = '1000',\n" +
                        "    'sink.buffer-flush.interval' = '5000',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.enable-delete' = 'true',\n" +
                        "    'sink.properties.format' = 'json',\n" +
                        "    'sink.enable.batch-mode' = 'true'\n" +
                        ")"
        );




/*
insert into dws_community_wide
SELECT
    -- 小区维度
    c.id AS community_id,
    c.communityName AS community_name,
    c.district_id,
    c.city_id,
    c.communityLocation AS community_location,
    c.remark AS community_remark,

    -- 楼栋维度
    h.id AS house_id,
    h.cid AS house_cid,
    h.houseCode AS house_code,
    h.houseName AS house_name,
    h.buildStartTime AS build_start_time,
    h.buildEndTime AS build_end_time,
    h.houseArea AS house_area,
    h.remark AS house_remark,

    -- 业主维度
    o.id AS ower_id,
    o.roomId AS ower_room_id,
    o.owerCode AS ower_code,
    o.owerName AS ower_name,
    o.owerSex AS ower_sex,
    o.owerAge AS ower_age,
    o.owerIdcard AS ower_idcard,
    o.owerTel AS ower_tel,
    o.owerHomeplace AS ower_homeplace,
    o.owerWorkplace AS ower_workplace,
    o.createTime AS ower_create_time,
    o.operatorTime AS ower_operator_time,
    o.remark AS ower_remark,

    -- 车辆维度
    cars.id AS car_id,
    cars.owerId AS car_ower_id,
    cars.carCode AS car_code,
    cars.carColor AS car_color,
    cars.`type` AS car_type,
    cars.remark AS car_remark
FROM ods_community c
-- 关联楼栋表（小区 -> 楼栋）
LEFT JOIN ods_house h ON c.id = h.cid
-- 关联业主表（楼栋 -> 业主）
LEFT JOIN ods_ower o ON h.id = o.roomId
-- 关联车辆表（业主 -> 车辆）
LEFT JOIN ods_cars cars ON o.id = cars.owerId;
 */     // 插入语句
        tEnv.executeSql(
                "insert into dws_community_wide\n" +
                        "SELECT\n" +
                        "    -- 小区维度\n" +
                        "    c.id AS community_id,\n" +
                        "    c.communityName AS community_name,\n" +
                        "    c.district_id,\n" +
                        "    c.city_id,\n" +
                        "    c.communityLocation AS community_location,\n" +
                        "    c.remark AS community_remark,\n" +
                        "\n" +
                        "    -- 楼栋维度\n" +
                        "    h.id AS house_id,\n" +
                        "    h.cid AS house_cid,\n" +
                        "    h.houseCode AS house_code,\n" +
                        "    h.houseName AS house_name,\n" +
                        "    h.buildStartTime AS build_start_time,\n" +
                        "    h.buildEndTime AS build_end_time,\n" +
                        "    h.houseArea AS house_area,\n" +
                        "    h.remark AS house_remark,\n" +
                        "\n" +
                        "    -- 业主维度\n" +
                        "    o.id AS ower_id,\n" +
                        "    o.roomId AS ower_room_id,\n" +
                        "    o.owerCode AS ower_code,\n" +
                        "    o.owerName AS ower_name,\n" +
                        "    o.owerSex AS ower_sex,\n" +
                        "    o.owerAge AS ower_age,\n" +
                        "    o.owerIdcard AS ower_idcard,\n" +
                        "    o.owerTel AS ower_tel,\n" +
                        "    o.owerHomeplace AS ower_homeplace,\n" +
                        "    o.owerWorkplace AS ower_workplace,\n" +
                        "    o.createTime AS ower_create_time,\n" +
                        "    o.operatorTime AS ower_operator_time,\n" +
                        "    o.remark AS ower_remark,\n" +
                        "\n" +
                        "    -- 车辆维度\n" +
                        "    cars.id AS car_id,\n" +
                        "    cars.owerId AS car_ower_id,\n" +
                        "    cars.carCode AS car_code,\n" +
                        "    cars.carColor AS car_color,\n" +
                        "    cars.`type` AS car_type,\n" +
                        "    cars.remark AS car_remark\n" +
                        "FROM ods_community c\n" +
                        "-- 关联楼栋表（小区 -> 楼栋）\n" +
                        "LEFT JOIN ods_house h ON c.id = h.cid\n" +
                        "-- 关联业主表（楼栋 -> 业主）\n" +
                        "LEFT JOIN ods_ower o ON h.id = o.roomId\n" +
                        "-- 关联车辆表（业主 -> 车辆）\n" +
                        "LEFT JOIN ods_cars cars ON o.id = cars.owerId;"
        );




/*
-- 统计累计指标
SELECT
    COUNT(DISTINCT community_id) AS cumulative_communities, -- 累计进驻小区
    COUNT(DISTINCT house_id) AS cumulative_buildings,       -- 累计楼栋
    COUNT(DISTINCT ower_id) AS cumulative_owners,           -- 累计业主
    COUNT(DISTINCT car_id) AS cumulative_cars                -- 累计车辆
FROM dws_community_wide;
 */
        //env.execute("PageOverallCount");
    }
}
